package com.xiao.rpc.registry.Registrys;


import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;

import com.xiao.rpc.config.RegistryConfig;
import com.xiao.rpc.model.ServiceMetaInfo;
import com.xiao.rpc.registry.Registry;

import com.xiao.rpc.registry.RegistryServiceMultiCache;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;



/**
 * etcd 注册中心
 */
@Slf4j
public class EtcdRegistry implements Registry {

    private Client client;

    private KV kvClient;

    /**
     * 本机注册的节点 key 集合（用于维护续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();
    /**
     * 注册中心服务缓存
     */
    private final RegistryServiceMultiCache registryServiceMultiCache = new RegistryServiceMultiCache();

    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();


    /**
     * 根节点
     *
     * 读取注册中心配置，并初始化客户端对象
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";

    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder()
                .endpoints(registryConfig.getRegistryAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        heartbeat();
    }

    /**
     * 服务注册
     * <p>
     * 服务注册，创建key并设置过期时间，value为服务注册信息的JSON序列化。
     *
     * @param serviceMetaInfo
     * @throws Exception
     */
    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // 创建Lease和KV客户端
        Lease leaseClient = client.getLeaseClient();
        // 创建一个30秒的租约
        long leaseId = leaseClient.grant(30).get().getID();

        // 设置要存储的键值对
        String registryKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registryKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        // 将键值对与租约绑定，并且设置过期时间
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, putOption).get();

        // 添加节点信息到本地缓存中
        localRegisterNodeKeySet.add(registryKey);
    }

    /**
     * 服务注销
     *
     * @param serviceMetaInfo
     */
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String registryKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.
                from(ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey(), StandardCharsets.UTF_8));
        // 删除节点信息从本地缓存中
        localRegisterNodeKeySet.remove(registryKey);
    }

    /**
     * 服务发现
     *
     * @param serviceKey
     * @return
     */
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 优先从缓存获取服务信息
        // 原教程————不支持多个服务同时缓存
        // List<ServiceMetaInfo> cachedServiceMetaInfoList = registryServiceCache.readCache();
        // 优化后————支持多个服务同时缓存
        List<ServiceMetaInfo> cachedServiceMetaInfoList = registryServiceMultiCache.readCache(serviceKey);

        if (cachedServiceMetaInfoList != null) {
            // 缓存生效测试：缓存命中日志
            log.info("从缓存获取服务列表，serviceKey: {}", serviceKey);
            return cachedServiceMetaInfoList;
        }

        // 前缀搜索，结尾一定要加 '/'
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";

        try {
            // 前缀查询
            GetOption getOption = GetOption.builder().isPrefix(true).build();
            List<KeyValue> keyValues = kvClient.get(
                            ByteSequence.from(searchPrefix, StandardCharsets.UTF_8),
                            getOption)
                    .get()
                    .getKvs();
            // 解析服务信息
            List<ServiceMetaInfo> serviceMetaInfoList = keyValues.stream()
                    .map(keyValue -> {
                        String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                        // 监听key 的变化
                        watch(key);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    })
                    .collect(Collectors.toList());
            // 写入服务缓存
            // 原教程代码，不支持多个服务同时缓存
            // registryServiceCache.writeCache(serviceMetaInfoList);
            // 优化后的代码，支持多个服务同时缓存
            registryServiceMultiCache.writeCache(serviceKey, serviceMetaInfoList);
            // 缓存生效测试：从注册中心获取并写入缓存的日志
            log.info("从注册中心获取服务列表并写入缓存，serviceKey: {}", serviceKey);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    /**
     * 服务中心销毁
     */
    public void destroy() {
        System.out.println("当前节点下线");
        // 下线节点
        // 遍历本节点所有的 key
        for (String key : localRegisterNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(key + "节点下线失败");
            }
        }

        // 释放资源
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }


    /**
     * 服务心跳
     *
     * 使用Hutool工具类的CronUtil实现定时任务，对所有集合中的节点执行重新注册操作，就相当于续签了。
     * 采用这种实现方案的好处: 即使Etcd注册中心的数据出现了丢失，通过心跳检测机制也会重新注册节点信息。
     */
    @Override
    public void heartbeat() {
        // 10秒续签一次
        CronUtil.schedule("0/10 * * * * ?", new Task() {
            @Override
            public void execute() {
                // 遍历本地注册所有的节点，进行心跳续期
                for (String key : localRegisterNodeKeySet) {
                    try {
                        // 1. 查询注册中心中该Key的记录（Etcd的KV查询）
                        List<KeyValue> keyValues = kvClient
                                .get(ByteSequence.from(key, StandardCharsets.UTF_8))
                                .get()
                                .getKvs();
                        // 2. 判断是否过期：Key不存在→已过期，跳过；
                        if (CollUtil.isEmpty(keyValues)) {
                            continue;
                        }
                        // 3. Key存在→未过期
                        // 续期：解析服务元信息，复用register方法重新注册（刷新租约）
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + "服务续签失败", e);
                    }
                }
            }
        });
        // 支持秒级别定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    /**
     * 监听节点变化
     *
     * @param serviceNodeKey
     */
    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        // 之前未被监听的，开启监听
        boolean newWhach = watchingKeySet.add(serviceNodeKey);
        if (newWhach) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        case DELETE:
                            // 清理注册缓存
                            // 原教程代码，不支持多个服务同时缓存
                            // registryServiceCache.clearCache();
                            // 优化后的代码，支持多个服务同时缓存
                            // fixme 这里需要改为 serviceKey，而不是 serviceNodeKey
                            registryServiceMultiCache.clearCache(serviceNodeKey);
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }
}
